/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.client.impl.consumer;

import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import com.alibaba.rocketmq.client.consumer.AllocateMessageQueueStrategy;
import com.alibaba.rocketmq.client.consumer.store.OffsetStore;
import com.alibaba.rocketmq.client.consumer.store.ReadOffsetType;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.impl.factory.MQClientInstance;
import com.alibaba.rocketmq.common.MixAll;
import com.alibaba.rocketmq.common.UtilAll;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageQueue;
import com.alibaba.rocketmq.common.protocol.heartbeat.ConsumeType;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;

/**
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-6-22
 */
public class RebalancePushImpl extends RebalanceImpl {
	private final DefaultMQPushConsumerImpl defaultMQPushConsumerImpl;

	public RebalancePushImpl(DefaultMQPushConsumerImpl defaultMQPushConsumerImpl) {
		this(null, null, null, null, defaultMQPushConsumerImpl);
	}

	public RebalancePushImpl(String consumerGroup, MessageModel messageModel, AllocateMessageQueueStrategy allocateMessageQueueStrategy, MQClientInstance mQClientFactory, DefaultMQPushConsumerImpl defaultMQPushConsumerImpl) {
		super(consumerGroup, messageModel, allocateMessageQueueStrategy, mQClientFactory);
		this.defaultMQPushConsumerImpl = defaultMQPushConsumerImpl;
	}

	@Override
	public void dispatchPullRequest(List<PullRequest> pullRequestList) {
		/**
		 * 遍历入参List<PullRequest>列表，以每个PullRequest对象为参数调用DefaultMQPushConsumerImpl.executePullRequestImmediately(PullRequest pullRequest)方法，
		 * 在该方法中只调用MQClientInstance.pullMessageService.executePullRequestImmediately(PullRequest pullRequest)方法，
		 * 将PullRequest对象放入PullMessageService服务线程的pullRequestQueue队列中；
		 * 在启动Consumer时启动了PullMessageService服务线程，
		 * 由该服务线程在后台监听pullRequestQueue队列的数据，
		 * 若队列中有数据到来则执行相应的业务逻辑
		 */
		for (PullRequest pullRequest : pullRequestList) {
			this.defaultMQPushConsumerImpl.executePullRequestImmediately(pullRequest);
			log.info("doRebalance, {}, add a new pull request {}", consumerGroup, pullRequest);
		}
	}

	@Override
	public long computePullFromWhere(MessageQueue mq) {
		long result = -1;
		// 1、获取DefaultMQPushConsumer.consumeFromWhere变量值，在应用层设置该值，默认值为CONSUME_FROM_LAST_OFFSET；
		final ConsumeFromWhere consumeFromWhere = this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer().getConsumeFromWhere();

		// 2、获取DefaultMQPushConsumerImpl.offsetStore对象，若为广播模式则该对象初始化为LocalFileOffsetStore；若为集群模式则该对象初始化为RemoteBrokerOffsetStore对象；
		final OffsetStore offsetStore = this.defaultMQPushConsumerImpl.getOffsetStore();
		switch (consumeFromWhere) {
			case CONSUME_FROM_LAST_OFFSET_AND_FROM_MIN_WHEN_BOOT_FIRST :
			case CONSUME_FROM_MIN_OFFSET :
			case CONSUME_FROM_MAX_OFFSET :
				// 3、若consumeFromWhere值等于CONSUME_FROM_LAST_OFFSET（表示一个新的订阅组第一次启动从队列的最后位置开始消费）。
			case CONSUME_FROM_LAST_OFFSET : {
				// pull 对于LocalFileOffsetStore对象，从本地加载offsets.json文件，然后获取该MessageQueue对象的offset值；
				// push
				long lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE);
				if (lastOffset >= 0) {
					result = lastOffset;
				}
				// First start,no offset
				else if (-1 == lastOffset) {
					// 若该offset值等于-1
					if (mq.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
						// 若topic是以"%RETRY%"开头，则返回0
						result = 0L;
					} else {
						try {
							// 若不是以"%RETRY%"开头，则向该MessageQueue对象的brokername所在Broker发送GET_MAX_OFFSET请求码，
							// 获取该MessageQueue对象的topic和queueId对应的逻辑队列的最大逻辑偏移量，并返回该最大offset值；
							result = this.mQClientFactory.getMQAdminImpl().maxOffset(mq);
						} catch (MQClientException e) {
							result = -1;
						}
					}
				} else {
					result = -1;
				}
				break;
			}
			// 表示一个新的订阅组第一次启动从队列的最前位置开始消费
			case CONSUME_FROM_FIRST_OFFSET : {
				// readOffset逻辑同CONSUME_FROM_LAST_OFFSET一样
				long lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE);
				if (lastOffset >= 0) {
					result = lastOffset;
				} else if (-1 == lastOffset) {
					result = 0L;
				} else {
					result = -1;
				}
				break;
			}
			// 表示一个新的订阅组第一次启动从指定时间点开始消费
			case CONSUME_FROM_TIMESTAMP : {
				long lastOffset = offsetStore.readOffset(mq, ReadOffsetType.READ_FROM_STORE);
				if (lastOffset >= 0) {
					result = lastOffset;
				} else if (-1 == lastOffset) {
					// 若topic是以"%RETRY%"开头
					if (mq.getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
						try {
							// 向该MessageQueue对象的brokername所在Broker发送GET_MAX_OFFSET请求码，获取该MessageQueue对象的topic和queueId对应的逻辑队列的最大逻辑偏移量，并返回该最大offset值
							result = this.mQClientFactory.getMQAdminImpl().maxOffset(mq);
						} catch (MQClientException e) {
							result = -1;
						}
					} else {
						try {
							// 若返回的offset值不等于-1
							long timestamp = UtilAll.parseDate(this.defaultMQPushConsumerImpl.getDefaultMQPushConsumer().getConsumeTimestamp(), UtilAll.yyyyMMddHHmmss).getTime();
							// 取DefaultMQPushConsumer.consumeTimestamp变量的时间戳，向该MessageQueue对象的brokername所在Broker发送SEARCH_OFFSET_BY_TIMESTAMP请求码，获取距离该时间戳最近的逻辑队列的offset值，并返回该offset值；
							result = this.mQClientFactory.getMQAdminImpl().searchOffset(mq, timestamp);
						} catch (MQClientException e) {
							result = -1;
						}
					}
				} else {
					result = -1;
				}
				break;
			}

			default :
				break;
		}

		return result;
	}

	@Override
	public void messageQueueChanged(String topic, Set<MessageQueue> mqAll, Set<MessageQueue> mqDivided) {
	}

	@Override
	public boolean removeUnnecessaryMessageQueue(MessageQueue mq, ProcessQueue pq) {
		// 广播模式LocalFileOffsetStore 该对象的persist方法没有处理逻辑
		// 集群模式RemoteBrokerOffsetStore 首先以入参MessageQueue对象为key值从RemoteBrokerOffsetStore.offsetTable: ConcurrentHashMap<MessageQueue, AtomicLong>变量中获取偏移量offset值，然后调用updateConsumeOffsetToBroker(MessageQueue mq, long offset)方法向Broker发送UPDATE_CONSUMER_OFFSET请求码，向Broker进行消费进度信息的同步；
		this.defaultMQPushConsumerImpl.getOffsetStore().persist(mq);

		// 广播模式LocalFileOffsetStore 该对象的removeOffset方法没有处理逻辑
		// 集群模式RemoteBrokerOffsetStore 将该MessageQueue对象的记录中从RemoteBrokerOffsetStore.offsetTable: ConcurrentHashMap<MessageQueue, AtomicLong>集合中删除
		this.defaultMQPushConsumerImpl.getOffsetStore().removeOffset(mq);

		// 若是顺序消费、并且是集群
		if (this.defaultMQPushConsumerImpl.isConsumeOrderly() && MessageModel.CLUSTERING.equals(this.defaultMQPushConsumerImpl.messageModel())) {
			try {
				// 尝试在1秒内获取到ProcessQueue.lockConsume:Lock的互斥锁
				/**
				 * 在此使用ProcessQueue.lockConsume互斥锁的原因是保证在发起解锁的过程中没有线程在消费该消息队列
				 * （在回调应用层的MessageListenerOrderly实现类的consumeMessage方法时会获取该ProcessQueue.lockConsume互斥锁，与解锁形成互斥逻辑）
				 */
				if (pq.getLockConsume().tryLock(1000, TimeUnit.MILLISECONDS)) {
					try {
						/**
						 * 若获取到互斥锁则向该MessageQueue对象的BrokerName下面的主用Broker发送UNLOCK_BATCH_MQ请求码请求对MessageQueue进行解锁操作，
						 * 采用oneway形式发送请求消息，发送成功则直接返回true，最后释放此互斥锁；
						 */
						this.unlock(mq, true);
						return true;
					} finally {
						pq.getLockConsume().unlock();
					}
				} else {
					log.warn("[WRONG]mq is consuming, so can not unlock it, {}. maybe hanged for a while, {}", //
							mq, //
							pq.getTryUnlockTimes());
					// 若未获取到该lockConsume的互斥锁，则打印试图解锁的次数并返回false
					pq.incTryUnlockTimes();
				}
			} catch (Exception e) {
				log.error("removeUnnecessaryMessageQueue Exception", e);
			}

			return false;
		}
		return true;
	}

	@Override
	public ConsumeType consumeType() {
		return ConsumeType.CONSUME_PASSIVELY;
	}
}
